package com.sunday.common.mq.rabbit.study.spring.study.e2_Sending_Messages;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.MessagePropertiesBuilder;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.annotation.RabbitListenerAnnotationBeanPostProcessor;
import org.springframework.amqp.rabbit.config.AbstractRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * @Description: demo程序
 */
@Slf4j
@RestController
@RequestMapping("/test")
public class MQApi {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    private static final String listenerId = "queue_id";

    /**
     * ====================================== RabbitTemplate ================================
     */

    @GetMapping("send")
    public void send() {

//        rabbitTemplate.setExchange();
//        rabbitTemplate.setRoutingKey();
//        rabbitTemplate.convertAndSend("foo");

        rabbitTemplate.convertAndSend(RabbitConfiguration.EXCHANGE_NAME, RabbitConfiguration.ROUTING_KEY_NAME, "foo");

        rabbitTemplate.convertAndSend(RabbitConfiguration.EXCHANGE_NAME, RabbitConfiguration.ROUTING_KEY_NAME, "foo", new CorrelationData("123456"));

        rabbitTemplate.convertAndSend(RabbitConfiguration.EXCHANGE_NAME, RabbitConfiguration.ROUTING_KEY_NAME, "foo", message -> {
            MessageProperties properties = message.getMessageProperties();
            properties.setPriority(1);
            return message;
        });

        rabbitTemplate.convertAndSend(RabbitConfiguration.EXCHANGE_NAME, RabbitConfiguration.ROUTING_KEY_NAME, "foo", message -> {
            MessageProperties properties = message.getMessageProperties();
            properties.setPriority(1);
            return message;
        }, new CorrelationData("654321"));

    }


    /**
     * 2023-06-05 14:21:21.345  INFO [647d7ee1575d2ba52458bce81c22f56c,2458bce81c22f56c] 26060 --- [io-16500-exec-9] c.s.c.m.r.s.spring.study.e1_Base.MQApi  [101] : CorrelationData [id=77fd11ad-67e8-416c-96ac-565079e9f48e]
     * 2023-06-05 14:21:21.348  INFO [647d7ee1575d2ba52458bce81c22f56c,8429386e023f53f3] 26060 --- [     queue_id-1] c.s.c.m.r.s.spring.study.e1_Base.MQApi  [180] : (Body:'foo' MessageProperties [headers={spring_listener_return_correlation=fa098a38-e41c-4d76-a6d2-07f59ca2c897, spring_returned_message_correlation=77fd11ad-67e8-416c-96ac-565079e9f48e}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=1, redelivered=false, receivedExchange=tracing, receivedRoutingKey=router, deliveryTag=3, consumerTag=amq.ctag-JqYv9FXYluxP3MOEA7bl4A, consumerQueue=queue])
     * 2023-06-05 14:21:21.350  INFO [647d7ee1575d2ba52458bce81c22f56c,2458bce81c22f56c] 26060 --- [io-16500-exec-9] c.s.c.m.r.s.spring.study.e1_Base.MQApi  [109] : Confirm [ack=true]
     * 2023-06-05 14:21:21.350  INFO [,] 26060 --- [nectionFactory6] c.s.c.m.r.s.s.s.RabbitBraveConfiguration[48] : correlationData : CorrelationData [id=77fd11ad-67e8-416c-96ac-565079e9f48e]
     * 2023-06-05 14:21:21.350  INFO [,] 26060 --- [nectionFactory6] c.s.c.m.r.s.s.s.RabbitBraveConfiguration[49] : ack : true
     * 2023-06-05 14:21:21.350  INFO [,] 26060 --- [nectionFactory6] c.s.c.m.r.s.s.s.RabbitBraveConfiguration[50] : cause : null
     */
    @GetMapping("publisher0")
    public void publisher0() {
        log.info("==============================================================================");
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        log.info("{}", correlationData);
        rabbitTemplate.convertAndSend(RabbitConfiguration.EXCHANGE_NAME, RabbitConfiguration.ROUTING_KEY_NAME, "foo", message -> {
            MessageProperties properties = message.getMessageProperties();
            properties.setPriority(1);
            return message;
        }, correlationData);
        try {
            CorrelationData.Confirm confirm = correlationData.getFuture().get(10, TimeUnit.SECONDS);
            log.info("{}", confirm);
        } catch (Exception e) {
            log.error(e.toString(), e);
        }
        log.info("==============================================================================");
    }

    /**
     * 2023-06-05 14:19:17.350  INFO [647d7e65c6277fc978c15c67a9f14dc4,78c15c67a9f14dc4] 26060 --- [io-16500-exec-2] c.s.c.m.r.s.spring.study.e1_Base.MQApi  [120] : CorrelationData [id=7db6ecc4-e2a0-4565-9823-3725328e6d9d]
     * 2023-06-05 14:19:17.353  INFO [647d7e65c6277fc978c15c67a9f14dc4,78c15c67a9f14dc4] 26060 --- [io-16500-exec-2] c.s.c.m.r.s.spring.study.e1_Base.MQApi  [132] : Confirm [ack=true]
     * 2023-06-05 14:19:17.353  INFO [,] 26060 --- [nectionFactory5] c.s.c.m.r.s.s.s.RabbitBraveConfiguration[53] : exchange : tracing
     * 2023-06-05 14:19:17.353  INFO [,] 26060 --- [nectionFactory5] c.s.c.m.r.s.s.s.RabbitBraveConfiguration[54] : routingKey : router1
     * 2023-06-05 14:19:17.353  INFO [,] 26060 --- [nectionFactory5] c.s.c.m.r.s.s.s.RabbitBraveConfiguration[55] : message : (Body:'foo' MessageProperties [headers={spring_returned_message_correlation=7db6ecc4-e2a0-4565-9823-3725328e6d9d, traceparent=00-647d7e65c6277fc978c15c67a9f14dc4-fd651a6592f0a5c8-00}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=1, deliveryTag=0])
     * 2023-06-05 14:19:17.353  INFO [,] 26060 --- [nectionFactory5] c.s.c.m.r.s.s.s.RabbitBraveConfiguration[56] : replyCode : 312
     * 2023-06-05 14:19:17.353  INFO [,] 26060 --- [nectionFactory5] c.s.c.m.r.s.s.s.RabbitBraveConfiguration[57] : replyText : NO_ROUTE
     * 2023-06-05 14:19:17.353  INFO [,] 26060 --- [nectionFactory4] c.s.c.m.r.s.s.s.RabbitBraveConfiguration[48] : correlationData : CorrelationData [id=7db6ecc4-e2a0-4565-9823-3725328e6d9d]
     * 2023-06-05 14:19:17.353  INFO [,] 26060 --- [nectionFactory4] c.s.c.m.r.s.s.s.RabbitBraveConfiguration[49] : ack : true
     * 2023-06-05 14:19:17.353  INFO [,] 26060 --- [nectionFactory4] c.s.c.m.r.s.s.s.RabbitBraveConfiguration[50] : cause : null
     */
    @GetMapping("publisher1")
    public void publisher1() {
        log.info("==============================================================================");
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        log.info("{}", correlationData);
        rabbitTemplate.convertAndSend(
                RabbitConfiguration.EXCHANGE_NAME,
                RabbitConfiguration.ROUTING_KEY_NAME + "1",
                "foo",
                message -> {
                    MessageProperties properties = message.getMessageProperties();
                    properties.setPriority(1);
                    return message;
                }, correlationData);
        try {
            CorrelationData.Confirm confirm = correlationData.getFuture().get(10, TimeUnit.SECONDS);
            log.info("{}", confirm);
        } catch (Exception e) {
            log.error(e.toString(), e);
        }
        log.info("==============================================================================");
    }

    /**
     * 2023-06-05 14:13:41.596  INFO [647d7d15e3b305bdf974066807fef21d,f974066807fef21d] 26060 --- [io-16500-exec-6] c.s.c.m.r.s.spring.study.e1_Base.MQApi  [143] : CorrelationData [id=1a884ed9-a455-45d1-92cf-b80539a3fcaf]
     * 2023-06-05 14:13:41.602 ERROR [,] 26060 --- [ 127.0.0.1:5672] o.s.a.r.c.CachingConnectionFactory      [742] : Shutdown Signal: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'tracing1' in vhost 'my_vhost', class-id=60, method-id=40)
     * 2023-06-05 14:13:41.602  INFO [647d7d15e3b305bdf974066807fef21d,f974066807fef21d] 26060 --- [io-16500-exec-6] c.s.c.m.r.s.spring.study.e1_Base.MQApi  [155] : Confirm [ack=false, reason=channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'tracing1' in vhost 'my_vhost', class-id=60, method-id=40)]
     * 2023-06-05 14:13:41.602  INFO [,] 26060 --- [nectionFactory3] c.s.c.m.r.s.s.s.RabbitBraveConfiguration[48] : correlationData : CorrelationData [id=1a884ed9-a455-45d1-92cf-b80539a3fcaf]
     * 2023-06-05 14:13:41.602  INFO [,] 26060 --- [nectionFactory3] c.s.c.m.r.s.s.s.RabbitBraveConfiguration[49] : ack : false
     * 2023-06-05 14:13:41.602  INFO [,] 26060 --- [nectionFactory3] c.s.c.m.r.s.s.s.RabbitBraveConfiguration[50] : cause : channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'tracing1' in vhost 'my_vhost', class-id=60, method-id=40)
     */
    @GetMapping("publisher2")
    public void publisher2() {
        log.info("==============================================================================");
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        log.info("{}", correlationData);
        rabbitTemplate.convertAndSend(
                RabbitConfiguration.EXCHANGE_NAME + "1",
                RabbitConfiguration.ROUTING_KEY_NAME,
                "foo",
                message -> {
                    MessageProperties properties = message.getMessageProperties();
                    properties.setPriority(1);
                    return message;
                }, correlationData);
        try {
            CorrelationData.Confirm confirm = correlationData.getFuture().get(10, TimeUnit.SECONDS);
            log.info("{}", confirm);
        } catch (Exception e) {
            log.error(e.toString(), e);
        }
        log.info("==============================================================================");
    }

    @GetMapping("publisher3")
    public void publisher3() {
        log.info("==============================================================================");
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        MessageProperties props = MessagePropertiesBuilder.newInstance()
                .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
                .setMessageId("123")
                .setHeader("bar", "baz")
                .build();
        Message message = MessageBuilder.withBody("foo".getBytes())
                .andProperties(props)
                .setCorrelationId(correlationData.getId())
//                .setCorrelationId(UUID.randomUUID().toString())
                .build();
        rabbitTemplate.convertAndSend(
                RabbitConfiguration.EXCHANGE_NAME,
                RabbitConfiguration.ROUTING_KEY_NAME,
                message
                // 当设置 correlationData 的时候，template.setConfirmCallback correlationData 才会存在值
                // 与 MessageBuilder 中的 setCorrelationId 并不是一个维度， message中是在 correlationId 属性
                , correlationData
        );
        log.info("==============================================================================");
    }

    /**
     * ====================================== RabbitListenerEndpointRegistry ================================
     * <p>
     * {@link org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer}
     * {@link AbstractRabbitListenerContainerFactory#createListenerContainer(RabbitListenerEndpoint)}
     * {@link RabbitListenerEndpointRegistry#createListenerContainer(RabbitListenerEndpoint, RabbitListenerContainerFactory)}
     * {@link RabbitListenerAnnotationBeanPostProcessor#postProcessAfterInitialization(Object, String)}
     * <p>
     * https://docs.spring.io/spring-amqp/docs/3.0.4/reference/html/#preface
     * 当应用程序配置了单个 ，就像默认情况下使用 Spring 引导自动配置一样，当连接被代理阻止时，应用程序将停止工作。
     * 当它被经纪人阻止时，它的任何客户都会停止工作。
     * 如果我们在同一个应用程序中有生产者和消费者，当生产者阻止连接（因为代理上不再有资源）并且消费者无法释放它们（因为连接被阻止）时，
     * 我们最终可能会陷入死锁。 为了缓解此问题，我们建议再有一个具有相同选项的单独实例 - 一个用于生产者，一个用于使用者。
     * 对于在使用者线程上执行的事务生成者，不可能单独，因为它们应重用与使用者事务关联的。
     * CachingConnectionFactory CachingConnectionFactory CachingConnectionFactoryChannel
     */
    @RabbitListener(
            id = listenerId,
            queues = {RabbitConfiguration.QUEUE_NAME},
            concurrency = "1-5"
    )
    public void taskPlanCheck(Message message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {
        log.info("message ------>{}", message);
        log.info("channel ------>{}", channel);
        log.info("tag ------>{}", tag);
    }

}